package com.yide.myworld.service.impl;

import com.alibaba.fastjson2.JSON;
import com.yide.myworld.domain.MqDataDto;
import com.yide.myworld.domain.MqFlowMiddle;
import com.yide.myworld.exception.Asserts;
import com.yide.myworld.mapper.MqFlowMiddleMapper;
import com.yide.myworld.service.RocketMQService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;

import java.sql.Timestamp;

/**
 * @auth: byebyebye
 * @date: 2023/3/7 4:51 PM
 * @description:
 */
@Slf4j
public class RocketMQServiceImpl implements RocketMQService {

    @Autowired
    private MqFlowMiddleMapper mqFlowMiddleMapper;

    @Autowired(required = false)
    private RocketMQTemplate rocketMQTemplate;


    @Override
    public void syncSend(String topic, MqDataDto mqDto) {
        SendResult sendResult = rocketMQTemplate.syncSend(topic, mqDto);
        log.info("消息发送成功！[{}],{}", JSON.toJSONString(mqDto), sendResult);
        MqFlowMiddle djsMqFlowMiddle = new MqFlowMiddle();
        djsMqFlowMiddle.setMessageId(mqDto.getMessageId());
        djsMqFlowMiddle.setTopic(topic);
        djsMqFlowMiddle.setMsgId(sendResult.getMsgId());
        djsMqFlowMiddle.setSendTime(new Timestamp(System.currentTimeMillis()));
        mqFlowMiddleMapper.insert(djsMqFlowMiddle);

    }

    @Override
    public void syncSend(String topic, Object object) {
        syncSend(topic, new MqDataDto(object));
    }

    @Override
    public void asyncSend(String topic, MqDataDto mqDto, long timeout, int delayLevel) {
        rocketMQTemplate.asyncSend(topic, MessageBuilder.withPayload(JSON.toJSONString(mqDto)).build(), new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                MqFlowMiddle djsMqFlowMiddle = new MqFlowMiddle();
                djsMqFlowMiddle.setMessageId(mqDto.getMessageId());
                djsMqFlowMiddle.setTopic(topic);
                djsMqFlowMiddle.setMsgId(sendResult.getMsgId());
                djsMqFlowMiddle.setSendTime(new Timestamp(System.currentTimeMillis()));
                mqFlowMiddleMapper.insert(djsMqFlowMiddle);
                log.info("mq消息发送成功：{}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("mq消息发送失败,topic为：" + topic, throwable);
                Asserts.fail(throwable.getMessage());
            }
        }, timeout, delayLevel);
    }
}
